package com.lkd.client.emq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONPath;
import com.lkd.client.config.EmqConfig;
import com.lkd.client.config.MsgType;
import com.lkd.client.emq.msg.*;
import com.lkd.client.pojo.*;
import com.lkd.client.service.DataProcessService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * emq回调类
 */
@Component
@Slf4j
public class EmqMsgCallback implements MqttCallbackExtended {
    @Autowired
    private EmqClient emqClient;

    @Autowired
    private  EmqConfig config;

    @Autowired
    private DataProcessService dataProcessService;

    static final Map<String, Double> threadLocal = new HashMap();

//    @Resource
//    VersionMapper versionMapper;

    @Override
    public void connectionLost(Throwable throwable) {
        log.info("emq connect lost",throwable);
        try {
           TimeUnit.SECONDS.sleep(10);
           //连接中断，从新建立连接
           emqClient.connect(this);
           //订阅服务器通知topic
            emqClient.subscribe(config.getSubscribeTopic(),this);
        } catch (Exception e) {
          log.info("订阅失败",e);
        }
    }

    /**
     *收到下推消息时回调
     *当一个主题发送了消息后，这个主题消息会保存到mosquito上，只要监听这个主题的服务启动就会看到此主题最新的一次主题消息
     *由于多个机器都会接收这个主题的消息，会导致消息的重复消费，需要考虑重复数据问题（集群的时候，考虑加上分布式锁）
     *在此部分需要将接收到的消息解析，数据的持久化操作，具体为流程为：先对收到的来自于网关的json字符串进行反序列化，将其转化为sensor实体类，编写SQL语句，实现写数据库
     * @param topic
     * @param mqttMessage
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        log.info("=================================收到消息，开始回调===========================================");
        CompletableFuture.runAsync(()->{//异步处理
            System.out.println("收消息topic:"+topic);
            String payload = new String(mqttMessage.getPayload());
            String msgType =  String.valueOf(JSONPath.read(payload, "$.type"));//从接收的消息中读出协议类型 JSONPath.read(payload, "$.type");//从接收的消息中读出协议类型
            log.info("************上报报文:msgType {} payload {}*************", msgType, payload);
            if(MsgType.photoReq.getType().equalsIgnoreCase(msgType)){//再把枚举中的协议类型与上面的msgType进行比较，然后调用相应的处理方法
                XunjieSanLinResponse linResponse = JSON.parseObject(payload, XunjieSanLinResponse.class);
                log.info("XunjieSanLinResponse 计算 开始 ,采集时间:{}", linResponse.getTime());
//                点位信息：
//                投加流量D4012
//                模拟投加流量(边缘计算)   D11000
//                原水流量D4004
//                投加单耗D4008
//                原液浓度D4024
//                稀释浓度D712
                VarList varList = linResponse.getVarList();
                if(varList == null ){
                    log.error("varList 为空");
                    return;
                }
                double tjll = varList.getTJLL();  //
                double ysll = varList.getYSLL();
                double tjdh = varList.getTJDH();
                double yynd = varList.getYYND();
                double xsnd = varList.getXSND();
                if(xsnd == 0 || yynd ==0){
                    log.error("varList采集流量原液浓度 和 稀释浓度 为0, 不计算");
                    return;
                }
                double MN_RQ_TJLL = varList.getMN_RQ_TJLL();
                double mnRqTjll = ysll*tjdh/(yynd*xsnd);

                log.info("XunjieSanLinResponse 原始读取投加流量PLC值:{} 计算结果:{}  对比:{}" ,tjll, mnRqTjll,
                        Double.doubleToLongBits(tjll) == Double.doubleToLongBits(mnRqTjll) );
                varList.setMN_RQ_TJLL(mnRqTjll);
                double rqTjll = 0;
                if(threadLocal!=null  && threadLocal.size() > 0){
                    rqTjll =   threadLocal.get("RQ_TJLL");
                }
                log.info("缓存读取rqTjll结果:{} 计算结果对比:{}" , rqTjll, Double.doubleToLongBits(rqTjll) == Double.doubleToLongBits(mnRqTjll));
                //数据变化才写入
                if(Double.doubleToLongBits(rqTjll) != Double.doubleToLongBits(mnRqTjll) ||
                    Double.doubleToLongBits(MN_RQ_TJLL) != Double.doubleToLongBits(mnRqTjll)){
                    log.info("XunjieSanLinResponse 不相等则同步到PCL, 保存结果到db, 网关暂时不操作");
                    //写入数据库
                    threadLocal.put("RQ_TJLL",mnRqTjll);
                    extracted(mnRqTjll);
                    log.info("XunjieSanLinResponse 保存结果到MQTT成功");
                }else{
                    log.info("缓存读取rqTjll结果与计算结果对比一样,本次不写入");
                }




            }
        });

    }

    /**
     * @param mnRqTjll
     * //                {
     * //                    "command":"set_value_of_RQ_TJLL",
     * //                        "value":55
     * //                }
     */
    private void extracted(double mnRqTjll) {
        SanlinReqData sanlinReqData = new SanlinReqData();
        sanlinReqData.setCommand("set_value_of_RQ_TJLL");
        sanlinReqData.setValue(mnRqTjll);
        log.info("XunjieSanLinResponse 发送到MQ 完成: /sys/A5302202307A0000188/update");
        dataProcessService.linResponse(sanlinReqData);
    }

    /**
     * 消息发布成功后回调
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.info("deliveryComplete---------发布消息成功" + iMqttDeliveryToken.isComplete());
        try {
            iMqttDeliveryToken.getMessage();
            iMqttDeliveryToken.getClient();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 连接成功后回调
     * @param b
     * @param s
     */
    @Override
    public void connectComplete(boolean b, String s) {
        System.out.println("===============连接成功====================");
//        if(!connectComplete){
//            return;
//        }
        //订阅服务器topic,及回调处理方法
        try {
            log.info("connectComplete subscribe {} ",config.getSubscribeTopic());
            emqClient.subscribe(config.getSubscribeTopic(),this);
//            Version verson= versionMapper.selectById(1);
//            //发布同步消息
            System.out.println("===============发布同步消息====================");
            VersionReq versionReq=new VersionReq();
            versionReq.setInnerCode(config.getInnerCode());//设备编码
            versionReq.setSn(System.nanoTime());//唯一标识
//            versionReq.getData().setChannelCfg(123);
//            versionReq.getData().setSkucfgVersion(123);
//            versionReq.getData().setGSensorX(1.2222223);
//            versionReq.getData().setGSensorY(3.2222223);
//            versionReq.getData().setGSensorZ(2.2222223);

            versionReq.getData().setGnss(new GNSS("GNSS",1,2,3,4,5,new Date()));
            versionReq.getData().setGsensor(new Gsensor(100.1,103.2,109.3));
            versionReq.getData().setStationInfo(new StationInfo(100,200,300,400));
            versionReq.getData().setTrainingInfo(new TrainingInfo("101","102","103","104","105","106"));
            String msg=JSON.toJSONString(versionReq);
            log.info("msg"+msg);
            emqClient.publish(MsgType.versionCfg.getType(), JSON.toJSONString(versionReq));
//            //检查是否未同步的出货信息
//            dataProcessService.checkVendoutOrder();
        } catch (MqttException e) {
            log.info("connectComplete",e);
        }
    }
}
